package com.cshuig.hadoop.mapreduce.phoneflow;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Partitioner;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by hogan on 2015/10/28.
 * partitioner(分区): 可以将数据分发到不同的reducer中执行, 结果也是在不同的分区中
 * 何时执行partitioner: 在Mapper执行后, Reducer执行前
 * 所有partitioner泛型参数是Mapper的K2, V2
 */
public class PhonePatitioner extends Partitioner<Text, DataCount> {

    private static Map<String, Integer> map = new HashMap<String, Integer>();
    static {
        map.put("134", 1);
        map.put("135", 1);
        map.put("136", 1);
        map.put("137", 1);
        map.put("138", 1);
        map.put("139", 1);
        map.put("182", 2);
        map.put("183", 2);
        map.put("150", 3);
        map.put("159", 3);
    }
    /**
     * 如果reducer只有一个, 那么partitioner基本失效, 结果只在一个文件中
     * @param text
     * @param dataCount
     * @param numPartitions reducer数量, 通过job设置
     * @return 分区号
     */
    @Override
    public int getPartition(Text text, DataCount dataCount, int numPartitions) {
        Integer pnum = map.get(text.toString().substring(0, 3));
        pnum = pnum == null ? 0 : pnum;
        return pnum;
    }

}
